package com.heima.behavior.service.impl;

import com.alibaba.fastjson.JSON;
import com.heima.behavior.dto.ReadBehaviorDto;
import com.heima.behavior.entity.ApReadBehavior;
import com.heima.behavior.entity.UpdateArticleMessage;
import com.heima.behavior.service.IReadService;
import com.heima.common.dto.ResponseResult;
import com.heima.common.entity.User;
import com.heima.common.util.AppThreadLocal;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import springfox.documentation.spring.web.json.Json;

import java.util.Date;

@Service
public class ReadService implements IReadService {
    @Autowired
    MongoTemplate template;
    /**
     * 阅读行为
     * @param dto
     * @return
     */
    @Override
    public ResponseResult Read(ReadBehaviorDto dto) {
//        创建阅读对象
        ApReadBehavior apReadBehavior = new ApReadBehavior();
        User user = AppThreadLocal.get();
        if (user != null){
            apReadBehavior.setUserId(user.getUserId());
        }else {
            apReadBehavior.setEquipmentId(dto.getEquipmentId());
        }
        apReadBehavior.setArticleId(dto.getArticleId());
        apReadBehavior.setCount(1);
        apReadBehavior.setReadDuration(dto.getReadDuration());
        apReadBehavior.setPercentage(dto.getPercentage());
        apReadBehavior.setLoadDuration(dto.getLoadDuration());
        apReadBehavior.setCreatedTime(new Date());

//        查询是否纯在
        Query query = new Query();
        query.addCriteria(Criteria.where("authorId").is(dto.getArticleId()));
        if (user != null){
            query.addCriteria(Criteria.where("userId").is(user.getUserId()));
        }else {
            query.addCriteria(Criteria.where("equipmentId").is(dto.getEquipmentId()));
        }
        ApReadBehavior one = template.findOne(query, ApReadBehavior.class);
//        判断是否诶空
        if (one == null){
            template.save(apReadBehavior);
        }else {
            Update update = new Update();
            update.inc("count");
            update.set("readDuration",dto.getReadDuration());
            update.set("percentage",dto.getPercentage());
            update.set("loadDuration",dto.getLoadDuration());
            update.set("updatedTime",new Date());
            template.updateFirst(query,update,ApReadBehavior.class);
        }
//        发送阅读行为
        sendMessage(dto);
        return ResponseResult.okResult();
    }
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    @Value("${topic.read_behavior}")
    private String ReadTopic;
    /**
     * 发送阅读行为
     * @param dto
     */
    private void sendMessage(ReadBehaviorDto dto) {
        UpdateArticleMessage updateArticleMessage = new UpdateArticleMessage();
        updateArticleMessage.setArticleId(dto.getArticleId());
        updateArticleMessage.setType(0);
        updateArticleMessage.setAdd(1);
        String json = JSON.toJSONString(updateArticleMessage);
        kafkaTemplate.send(ReadTopic, json);
    }
}
